package ddbs;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/*
 * 
 */
public class PeerMgr implements MSGCallback, Runnable {
    static Map<String, MSGCallback> MSGCallbacks = new HashMap<String, MSGCallback>();
    private final String msgType = "ConfirmClientId";
    private Peer[] peer;
    private int id;
    private Selector selector;
    private SocketChannel currentChannel;
    Sched sch = null;

    int getSelfId(){
        return this.id;
    }
    public PeerMgr(Configuration conf) {
        this.peer = new Peer[4];
        this.id = Integer.parseInt(conf.getConf("id"));
        System.out.println("PeerId is: " + id);
        initServer(Integer.parseInt(conf.getConf("port" + id)));
        for (int i = 0; i != 4; ++i) {
            peer[i] = new Peer(i);
            peer[i].manager = this;
            peer[i].serverIp = conf.getConf("ip" + id);
            peer[i].serverPort = Integer.parseInt(conf.getConf("port" + i));
            connectPeer(getPeer(i));
        }
        registerCallback("ConfirmClientId", this);
    }

    public Peer getPeer(int id) {
        if (id >= 0 && id < 4)
            return this.peer[id];
        else {
            System.out.println("ERROR :here should never be reached");
            return null;
        }
    }

    public Peer getSelf() {
        return getPeer(this.id);
    }

    protected void initServer(int port) {
        try {
            ServerSocketChannel serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);
            serverChannel.socket().bind(new InetSocketAddress(port));
            selector = Selector.open();
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void handleMsg(Peer from, MSG msg) {
        int remoteId;
        remoteId = msg.id;
        Peer remote = getPeer(remoteId);
        System.out.println("registered client: " + remoteId);
        try {
            currentChannel.register(selector, SelectionKey.OP_READ, remote);
        } catch (ClosedChannelException e) {
            e.printStackTrace();
        }
        remote.toPeerChannel = currentChannel;
        if (remote.setConnected(true)) {
            remote.toPeerChannel = currentChannel;
        } else {
            try {
                currentChannel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    protected void connectPeer(Peer peer) {
        if (peer == null || peer.getConnected())
            return;
        if (this.id >= peer.getId())
            return;
        try {
            SocketChannel channel = SocketChannel.open();
            channel.configureBlocking(false);
            channel.connect(new InetSocketAddress(peer.serverIp,
                    peer.serverPort));
            channel.register(selector, SelectionKey.OP_CONNECT, peer);
            peer.toPeerChannel = channel;
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private ByteBuffer readBytes(SocketChannel channel, int length) throws IOException {
        ByteBuffer tempBuffer = ByteBuffer.allocate(length);
        tempBuffer.clear();
        //System.out.println("readBytes="+length);
        while (tempBuffer.remaining()>0) {
            int ret = channel.read(tempBuffer);
            //System.out.println("Read len="+ret+" remaining="+tempBuffer.remaining());
        }
        return tempBuffer;
    }

    private MSG readMSG(SelectionKey key) {
        Peer remote = (Peer) key.attachment();
        SocketChannel channel = (SocketChannel) key.channel();
        currentChannel = channel;
        ByteBuffer headBuffer = ByteBuffer.allocate(4);
        int tempRead, contentSize;
        try {
            tempRead = channel.read(headBuffer);
            if (tempRead == 0)
                return null;;
            contentSize = Peer.byte2Int(headBuffer.array());
            ByteBuffer tempBuffer = readBytes(channel, contentSize);
            String msg = new String(tempBuffer.array());
            /*
            int length = msg.length();
            if (length > 100)        length = 100;
            String sub = msg.substring(0, length);
            System.out.println("Recieve msg from "+(remote==null?"null":remote.getId()) 
                    +" length: " + msg.length()+" /"+length
                    + " msg is: [" + sub +"]");
            */
            MSG m = new MSG(msg.length(), msg);
            //m.show();
            return m;
        } catch (Exception e) {
            try {
                channel.close();
            } catch (IOException e1) {
            }
            key.cancel();
            System.out.println("Read Msg Error! peer=" + (remote == null ? ""
                    : remote.getId()));
             e.printStackTrace();
            if (remote != null){
                remote.setConnected(false);
                connectPeer(remote);
            }
            return null;
        }
    }

    public static void registerCallback(String type, MSGCallback cb) {
        MSGCallbacks.put(type, cb);
    }

    private void readCB(SelectionKey key) {
        MSG msg = null;
        Peer p = (Peer) key.attachment();
        while ((msg = readMSG(key)) != null) {
            if (p !=null) {
                p.transferSizeR +=msg.len;
            }
            MSGCallbacks.get(msg.type).handleMsg(p, msg);
        }
    }

    public void listen() throws IOException {
        System.out.println("Peer " + id + " listen on port "
                + getPeer(id).serverPort + " for incoming connections!");
        while (true) {
            selector.select();
            Iterator<SelectionKey> ite = selector.selectedKeys().iterator();
            while (ite.hasNext()) {
                // System.out.println("Peer " + id + " selector selecting!");
                SelectionKey key = (SelectionKey) ite.next();
                ite.remove();
                if (key.isAcceptable()) {
                    System.out.println("Peer " + this.id + " selector accepting!");
                    ServerSocketChannel server = (ServerSocketChannel) key
                            .channel();
                    SocketChannel channel = server.accept();
                    channel.configureBlocking(false);
                    channel.register(selector, SelectionKey.OP_READ, null);
                } else if (key.isConnectable()) {
                    Peer remote = (Peer) key.attachment();
                    SocketChannel channel = (SocketChannel) key.channel();
                    if (channel.isConnectionPending()) {
                        try {
                            channel.finishConnect();
                        } catch (IOException e) {
                            // e.printStackTrace();
                            channel.close();
                            key.cancel();
                            remote.setConnected(false);
                            connectPeer(remote);
                            continue;
                        }

                        if (!remote.setConnected(true)) {
                            channel.close();
                            key.cancel();
                            System.out.println("Peer " + this.id
                                    + "cancel connecting Peer "
                                    + remote.getId()
                                    + " while it is already connected");
                        } else {
                            channel.configureBlocking(false);
                            channel.register(selector, SelectionKey.OP_READ,
                                    remote);
                            MSG msg = new MSG(msgType, "", this.id, 0, "");
                            remote.sendMsg(msg);
                            System.out.println("Peer " + this.id
                                    + " finish connecting Peer "
                                    + remote.getId());
                        }
                    }
                } else if (key.isReadable()) {
                    readCB(key);
                }
            }
        }
    }

    /*
     * Load peer configurations and setup four peers:self and the three others
     * Start listening to a port and connect to the three others
     */
    public void run() {
        /*
         * New Threads: #1: Listen socket -- socket block/non-block
         */
        try {
            listen();
            System.out.println("listen halted?");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void start() {
        Thread t = new Thread(this);
        t.start();
    }
}
